package com.wyz.socket.core.impl;


import com.wyz.socket.core.api.Close;
import com.wyz.socket.core.api.Receiver;
import com.wyz.socket.core.api.Sender;
import com.wyz.socket.core.base.Context;
import com.wyz.socket.core.base.packet.AbstractReceivePacket;
import com.wyz.socket.core.base.packet.AbstractSendPacket;
import com.wyz.socket.core.base.packet.bytes.BytesReceivePacket;
import com.wyz.socket.core.base.packet.file.FileReceivePacket;
import com.wyz.socket.core.base.packet.stream.StreamDirectReceivePacket;
import com.wyz.socket.core.base.packet.string.StringReceivePacket;
import com.wyz.socket.core.base.packet.string.StringSendPacket;
import com.wyz.socket.core.impl.dispatcher.bridge.BridgeSocketDispatcher;
import com.wyz.socket.core.impl.chain.AbstractCloseChain;
import com.wyz.socket.core.impl.chain.AbstractStringPacketChain;
import com.wyz.socket.core.impl.chain.DefaultNonStringPacketChain;
import com.wyz.socket.core.impl.chain.DefaultPrintConnectorCloseChain;
import com.wyz.socket.core.impl.dispatcher.AsyncReceiveDispatcher;
import com.wyz.socket.core.impl.dispatcher.AsyncSendDispatcher;
import com.wyz.socket.core.api.dispatcher.ReceiveDispatcher;
import com.wyz.socket.core.api.dispatcher.SendDispatcher;
import com.wyz.socket.core.impl.scheduler.SchedulerJob;
import com.wyz.socket.core.utils.Log;

import java.io.File;
import java.io.IOException;
import java.io.OutputStream;
import java.net.StandardSocketOptions;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

import static com.wyz.socket.core.base.packet.Packet.*;
import static com.wyz.socket.core.base.packet.Packet.TYPE_STREAM_DIRECT;

public abstract class Connector implements Close {

    public final UUID key;
    public final String clientInfo;

    protected final AbstractCloseChain closeChain = new DefaultPrintConnectorCloseChain(); // 关闭责任链
    protected final AbstractStringPacketChain printChain = new DefaultNonStringPacketChain();     // 打印责任链

    private SendDispatcher sendDispatcher;
    private ReceiveDispatcher receiveDispatcher;
    private SocketChannelAdapter adapter;
    private final List<SchedulerJob> schedulerJobs = new ArrayList<>(4);

    private final ReceiveDispatcher.ReceiveCallback receiveCallback = new ReceiveDispatcher.ReceiveCallback() {
        @Override
        public void onReceivePacketComplated(AbstractReceivePacket<?, ?> packet, boolean isSucceed) {
            if (isSucceed) {
                onReceivedPacket(packet);
            }
        }

        @Override
        public AbstractReceivePacket<?, ?> onArriyedNewPacket(byte type, int length, byte[] headerInfo) {
            switch (type) {
                case TYPE_MEMORY_BYTES: {
                    return new BytesReceivePacket(length);
                }
                case TYPE_MEMORY_STRING: {
                    return new StringReceivePacket(length);
                }
                case TYPE_STREAM_FILE: {
                    return new FileReceivePacket(createFile(length, headerInfo), length);
                }
                case TYPE_STREAM_DIRECT: {
                    return new StreamDirectReceivePacket(createOutputStream(length, headerInfo), length);
                }
                default:
                    throw new UnsupportedOperationException("unSupported packet type : " + type);
            }
        }

        @Override
        public void onReceivedHeartbeat() {
            Log.printlnInfo("ConnectorHandler", "心跳包 : " + getLastActiveTime());
        }
    };

    protected Connector(String clientInfo) {
        this.clientInfo = clientInfo;
        key = UUID.randomUUID();
    }

    public void onStart(SocketChannel socketChannel) throws IOException {
        socketChannel.configureBlocking(false);
        socketChannel.socket().setSoTimeout(1000);
        socketChannel.socket().setPerformancePreferences(1, 3, 2);
        socketChannel.setOption(StandardSocketOptions.SO_RCVBUF, 16 * 1024);
        socketChannel.setOption(StandardSocketOptions.SO_SNDBUF, 16 * 1024);
        socketChannel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
        socketChannel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
        adapter = new SocketChannelAdapter(socketChannel, Context.get().getIoProvider() , this);
        sendDispatcher = new AsyncSendDispatcher(adapter);
        receiveDispatcher = new AsyncReceiveDispatcher(adapter, receiveCallback);
        receiveDispatcher.start();
    }

    /**
     * 改变socket模式为桥接模式
     * 普通模式是 (Client-A)Sender[Receiver] > SendPacket    > Frame > Ioargs
     * .................................................................//
     * .........(Server)Receiver[Sender]    > ReceivePacket > Frame > Ioargs
     * ...........(分析|转发)
     * .........(Server)Sender[Receiver]    > SendPacket    > Frame > Ioargs
     * .................................................................//
     * .......(Client-B)Receiver[Sender]    > SendPacket    > Frame > Ioargs
     * <p>
     * 桥接模式是 [Client-A]Sender(Receiver) > SendPacket    > Frame > Ioargs
     * .................................................................//
     * ......... [Client-A]Receiver(Sender)  > ReceivePacket > Frame > Ioargs
     */
    public void changeSocketModeToBridge() {
        if (receiveDispatcher instanceof BridgeSocketDispatcher) {
            return;
        }
        receiveDispatcher.stop(); // close
        this.receiveDispatcher = new BridgeSocketDispatcher(adapter);
        this.receiveDispatcher.start();
    }

    public void bindToBridge(Sender sender) {
        if (adapter == sender) {
            throw new UnsupportedOperationException("can not set current connector sender");
        }
        if (!(receiveDispatcher instanceof BridgeSocketDispatcher)) {
            throw new UnsupportedOperationException("receiveDispatcher is not bridgeSocketdispatcher");
        }
        ((BridgeSocketDispatcher) receiveDispatcher).bindSender(sender);
    }

    public void unBindBridge() {
        if (!(receiveDispatcher instanceof BridgeSocketDispatcher)) {
            throw new UnsupportedOperationException("receiveDispatcher is not bridgeSocketdispatcher");
        }
        ((BridgeSocketDispatcher) receiveDispatcher).bindSender(null);
    }

    public Sender sender() {
        return adapter;
    }

    public Receiver receiver() {
        return adapter;
    }

    // abstract func
    protected abstract void onReceivedPacket(AbstractReceivePacket<?, ?> packet);

    protected abstract File createFile(int length, byte[] headerInfo);

    protected abstract OutputStream createOutputStream(int length, byte[] headerInfo);

    // send packet

    public void send(String msg) {
        StringSendPacket stringSendPacket = new StringSendPacket(msg);
        sendDispatcher.send(stringSendPacket);
    }

    public void send(AbstractSendPacket<?> packet) {
        sendDispatcher.send(packet);
    }


    // func
    public long getLastActiveTime() {
        return Math.max(sendDispatcher.getLastActiveTime(), receiveDispatcher.getLastActiveTime());
    }

    public void timeoutEvent() {
        sendDispatcher.sendHeartbeat();
    }

    public void timeoutEventException(Throwable throwable) {

    }

    public void schedule(SchedulerJob job) {
        synchronized (schedulerJobs) {
            if (schedulerJobs.contains(job)) {
                return;
            }
            job.scheduler(Context.get().getScheduler());
            schedulerJobs.add(job);
        }
    }

    // final close
    @Override
    public void close() {
        closeHeartBeat();
        adapter.close();
    }

    private void closeHeartBeat() {
        SchedulerJob[] schedulerJobs;
        synchronized (this.schedulerJobs) {
            schedulerJobs = this.schedulerJobs.toArray(new SchedulerJob[]{});
            this.schedulerJobs.clear();
        }
        for (SchedulerJob job : schedulerJobs) {
            job.unScheduler();
        }
    }
}
